package com.junxonline.instant.job.zookeeper;

import com.junxonline.instant.job.config.ZookeeperConfig;
import lombok.extern.log4j.Log4j2;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetAddress;

/**
 * Zookeeper Client
 *
 * @author JunX
 * @date 2021-06-17
 */
@Log4j2
@Component
public class ZookeeperClient {

    @Autowired
    private ZookeeperConfig zookeeperConfig;

    /**
     * Zookeeper客户端
     */
    private CuratorFramework zookeeperClient;

    /**
     * Zookeeper Client启动
     * @throws Exception Exception
     */
    @PostConstruct
    public void start() throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperConfig.getBaseSleepTimeMs(), zookeeperConfig.getMaxRetries());
        zookeeperClient = CuratorFrameworkFactory
                .builder()
                .connectString(zookeeperConfig.getZookeeperAddress())
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMs()).connectionTimeoutMs(zookeeperConfig.getConnectionTimeoutMs()).build();
        zookeeperClient.start();

        String name = zookeeperConfig.getId();
        String zookeeperAddress = zookeeperConfig.getZookeeperAddress();
        String ipPort = InetAddress.getLocalHost().getHostAddress() + ":" + zookeeperConfig.getPort();
        String data = name + "#" + zookeeperAddress + "#" + ipPort;
        zookeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zookeeperConfig.getRegistryPath() + "/" + zookeeperConfig.getId() + "-", data.getBytes());
        log.info("- Zookeeper client started on -> {} in path: {}", ipPort, zookeeperConfig.getRegistryPath());
    }

    /**
     * Zookeeper Client停止
     */
    @PreDestroy
    public void stop() {
        zookeeperClient.close();
        log.info("Zookeeper client stopped on port:{}", zookeeperConfig.getPort());
    }

}
